use std::{
    fs, io,
    net::{TcpListener, TcpStream}, thread,
    sync::{
        atomic::*, mpsc::{self, Receiver}, Arc, LazyLock,
    },
};
use tungstenite::{Message, WebSocket};
use crate::{
    defines::*, log, config::Config, utils::{self, lock_free_dt::IndexedList},
    protocol::{self, ProtocolIO}, engine,
};

const FORCE_QUIT_FILENAME:&str = "./QUIT_SIGNAL";

fn fuck_socket_read(ws:&mut WebSocket<TcpStream>) -> Option<Result<Message, tungstenite::Error>> {
    debug_assert!(ws.can_read());
    match ws.read() {
        Ok(a) => {Some(Ok(a))}
        Err(e) => {
            match e {
                tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed  => {None}
                tungstenite::Error::Io(ref e) if e.kind() == io::ErrorKind::WouldBlock => {None}
                _ => {Some(Err(e))}
            }
        }
    }
}

fn fuck_accept(stream:TcpStream) -> Result<WebSocket<TcpStream>, tungstenite::HandshakeError<tungstenite::ServerHandshake<TcpStream, tungstenite::handshake::server::NoCallback>>> {
    fn fuck_handshake(fucker:tungstenite::handshake::MidHandshake<tungstenite::ServerHandshake<TcpStream, tungstenite::handshake::server::NoCallback>>) ->
        Result<WebSocket<TcpStream>, tungstenite::HandshakeError<tungstenite::ServerHandshake<TcpStream, tungstenite::handshake::server::NoCallback>>> {
        match fucker.handshake() {
            Ok(stream) => {Ok(stream)}
            Err(e) => {
                match e {
                    tungstenite::HandshakeError::Interrupted(fucker) => {
                        fuck_handshake(fucker)
                    }
                    _ => {Err(e)}
                }
            }
        }
    }
    match tungstenite::accept(stream) {
        Ok(stream) => {Ok(stream)}
        Err(e) => {
            match e {
                tungstenite::HandshakeError::Interrupted(fucker) => {fuck_handshake(fucker)}
                _ => {Err(e)}
            }
        }
    }
}

fn socket_reply(ws:&mut WebSocket<TcpStream>, text:String) {
    debug_assert!(ws.can_write());
    if let Err(e) = ws.send(Message::text(text)) {
        log::warning!("[socket-reply] {}", e)
    }
}
fn socket_close(ws:&mut WebSocket<TcpStream>, text:Option<String>) {
    if let Some(text) = text {
        socket_reply(ws, text);
    }
    let _ = ws.send(Message::Close(None));
}
fn socket_usable(ws:&WebSocket<TcpStream>) -> bool {
    ws.can_read() && ws.can_write()
}

fn socket_reader(cid:ClientId, ws:&mut WebSocket<TcpStream>) {
    if let Some(fucked) = fuck_socket_read(ws) {
        match fucked {
            Ok(msg) => {
                if msg.is_binary() || msg.is_text() {
                    if let Ok(text) = msg.to_text() {
                        if let Err(e) = protocol::do_message(cid, text) {
                            log::warning!("[socket|client:{}] protocol message error: {}", cid, e);
                        }
                    } else {
                        socket_reply(ws, protocol::reply::error(&format!("error|not text message")));
                    }
                }
            },
            Err(e) => {
                socket_close(ws, None);
                log::error!("[socket|client:{}] eat a error with in reading: {}", cid, e);
            },
        }
    }
}

struct Client {
    socket: WebSocket<TcpStream>,
    rx: Receiver<ProtocolIO>,
}
unsafe impl Send for Client {}
unsafe impl Sync for Client {}
static CLIENTS:LazyLock<IndexedList<Client>> = LazyLock::new(||{IndexedList::new(CLIENT_MAX_NUMS)});

fn session_start(mut ws:WebSocket<TcpStream>, n_links:&AtomicI32) -> Option<(Error, WebSocket<TcpStream>)> {
    let (tx,rx) = mpsc::channel::<ProtocolIO>();
    match protocol::alloc(tx.clone()) {
        Ok(cid) => {
            debug_assert!(CLIENTS.get(cid).is_none());
            let _ = CLIENTS.set(cid, Client {
                socket: ws, rx,
            });
            n_links.fetch_add(1, Ordering::SeqCst);
            log::info!("[session|client:{}] begin", cid);
            None
        }
        Err(e) => {
            socket_reply(&mut ws, protocol::reply::error(&format!("protocol can't alloc: {}", e)));
            Some((anyhow!("protocol can't alloc: {}", e), ws))
        }
    }
}
fn session_end(cid:ClientId, n_links:&AtomicI32) {
    let mut slot = CLIENTS.remove(cid).unwrap();
    socket_close(&mut slot.socket, None);
    protocol::free(cid);
    n_links.fetch_sub(1, Ordering::SeqCst);
    log::info!("[session|client:{}] end", cid);
}

fn protocol_reciver(cid:ClientId, running:&AtomicBool, client:&mut Client) {
    let ws = &mut client.socket;
    match client.rx.try_recv() {
        Ok(pio) => {
            match pio {
                ProtocolIO::Reply(a) => {socket_reply(ws, a);}
                ProtocolIO::Close(a) => {
                    log::warning!("[protocol-reciver|client:{}] protocol request close session", cid);
                    socket_close(ws, Some(a));
                }
                ProtocolIO::Quit => {
                    log::info!("[protocol-reciver|client:{}] take quit message", cid);
                    running.store(false, Ordering::Relaxed);
                }
            }
        }
        Err(mpsc::TryRecvError::Empty) => {}
        Err(mpsc::TryRecvError::Disconnected) => {
            log::error!("[protocol-reciver|client:{}] protocol-channel has unlink", cid);
            socket_close(ws, Some(protocol::reply::error("engine crash: internal error")));
        }
    }
}

fn make_server_io(n_links:Arc<AtomicI32>, running:Arc<AtomicBool>) -> impl Fn() {
    move || {
        log::info!("[server-io] start");
        let mut frame_sync = utils::FrameSync::with_ms(SERVER_IO_MS);
        while running.load(Ordering::Relaxed) {
            for i in 0..CLIENT_MAX_NUMS {
                if let Some(client) = CLIENTS.get_mut(i) {
                    if socket_usable(&client.socket) {
                        socket_reader(i, &mut client.socket);
                    }
                    if socket_usable(&client.socket) {
                        protocol_reciver(i, &running, client);
                    }
                    if ! (socket_usable(&client.socket) && protocol::instance_is_alive(i)) {
                        session_end(i, &n_links);
                    }
                }
            }
            frame_sync.sync();
        }
        log::info!("[server-io] end");
    }
}

fn check_force_quit() -> Result<bool> {
    let have_signal = fs::exists(FORCE_QUIT_FILENAME)?;
    if have_signal {
        fs::remove_file(FORCE_QUIT_FILENAME)?;
        log::info!("[listener] server will force quit");
        Ok(true)
    } else {
        Ok(false)
    }
}

fn server_main(listener:TcpListener) {
    let running:Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
    let n_links:Arc<AtomicI32> = Arc::new(AtomicI32::new(0));

    let io_handle = thread::spawn(make_server_io(n_links.clone(), running.clone()));

    let mut frame_sync = utils::FrameSync::with_ms(SERVER_LISTENER_MS);
    for stream in listener.incoming() {
        if check_force_quit().unwrap_or(false) {
            running.store(false, Ordering::Relaxed);
            protocol::end();
            engine::end();
        }
        if ! running.load(Ordering::Relaxed) {
            log::info!("[listener] take quit signal");
            break;
        }
        match stream {
            Ok(stream) => {
                log::info!("[listener] make an stream");
                match fuck_accept(stream)  {
                    Ok(mut ws) => {
                        let link_id = n_links.load(Ordering::SeqCst);
                        log::info!("[listener|link:{}] has accept", link_id);
                        if link_id >= CLIENT_MAX_NUMS as i32 {
                            socket_close(&mut ws, Some(protocol::reply::error(&format!("over max client connect-nums:{}", CLIENT_MAX_NUMS))));
                        } else if link_id < 0 {
                            socket_close(&mut ws, Some(protocol::reply::error("internal error")));
                        } else {
                            if let Some((e, mut ws)) = session_start(ws, &n_links) {
                                socket_close(&mut ws, Some(protocol::reply::error("session create failure")));
                                log::error!("[listener|link:{}] session create failure: {}", link_id, e);
                            }
                        }
                    }
                    Err(e) => {
                        log::error!("[listener] accept failure: {}", e);
                    }
                }
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                log::error!("[listener] error: {}", e);
            }
        }
        frame_sync.sync();
    }
    if let Err(_) = io_handle.join() {
        log::warning!("[server-io] thread incorrect fuck")
    }
}


pub fn start(config:&Config) -> Result<()> {
    let addr = format!("localhost:{}", config.port);
    if let Ok(listener) = TcpListener::bind(&addr) {
        log::info!("[server] bind to port {}", config.port);
        if let Ok(()) = listener.set_nonblocking(true) {
            log::info!("[listener] has non-block");
            if let Ok(e) = listener.take_error() {
                if e.is_some() {
                    let e = e.unwrap();
                    log::error!("[server] can't start: {}", e);
                    return Err(e.into())
                }
            }
            log::info!("[server] start");
            server_main(listener);
            log::info!("[server] end");
            Ok(())
        } else {
            let e = anyhow!("[listener] can't set to non-block");
            log::error!("{}", e);
            Err(e)
        }
    } else {
        let e = anyhow!("[server] can't bind to port {}", config.port);
        log::error!("{}", e);
        Err(e)
    }
}

pub fn force_quit() -> Result<()> {
    if fs::exists(FORCE_QUIT_FILENAME)? {
        Ok(())
    } else {
        Ok(fs::write(FORCE_QUIT_FILENAME, b"SSLASH")?)
    }
}

